#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <errno.h>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <arpa/inet.h>

#include <ZLog.h>
#include <ZList.h>
#include <ZAtomic.h>
#include <ZQueue.h>
#include <ZMemPool.h>

#include <HTTPLiveProxy.h>

#define HL_LISTEN_PORT 4670
#define PERF_LIMIT 1024
#define CRLF "\r\n"

static const char *response200 = "HTTP/1.0 200 OK"CRLF
                                 "Server: Light-HTTP"CRLF
                                 "Content-Type: application/octet-stream"CRLF
                                 "Transfer-Encoding: chunked"CRLF;
                               //  CRLF; /* will add in chunked buffer */
                                 
static const char *response404 = "HTTP/1.0 404 Not Found"CRLF
                                 CRLF;
                                 
static const char *response503 = "HTTP/1.0 503 Not Overall Server"CRLF
                                 CRLF;

#define zsk_nonblocking(sk)  fcntl(sk, F_SETFL, fcntl(sk, F_GETFL) | O_NONBLOCK)

enum HLStateE
{
//    HL_STATE_INIT,
    HL_STATE_RUNNING,
    HL_STATE_STOPED /* no thread, all queue and list empty, can not fill */
};

enum HLSessionStateE
{
    HLSESSION_STATE_CONNECTED, /* a new session should be connected */
    HLSESSION_STATE_RECVHDR,
    HLSESSION_STATE_RESPONE_SEND,
    HLSESSION_STATE_TRANSFER,
};

enum HLMsgTypeE
{   
    HL_MSG_FILL,
    HL_MSG_LISTEN,
//    HL_MSG_SESSION_TRANSACTION,
    HL_MSG_STOP,
    
};

typedef struct HLSessionS HLSessionT;
typedef struct HLBufferS HLBufferT;
typedef struct HLBufferNodeS HLBufferNodeT;
typedef struct HLMessageS HLMessageT;

struct HLSessionS
{
    enum HLSessionStateE state;
    int sk;
    ZListT bufList;
    int bufPos;
    ZListNodeT node; /* in session list */

    char *hdr;
    int hdrLen;
    int hdrValid;
    int pathValid;

    const char *response;
    int responseLen;
    int responseSend;
};

struct HLBufferS
{
    char *data;
    int len;
    z_atomic_t ref;
};

struct HLBufferNodeS
{
    HLBufferT *buf;
    int offset;
    ZListNodeT node; /* in impl.bufList or session.bufList */
};

struct HLMessageS
{
    enum HLMsgTypeE type;
    union
    {
        HLBufferT *buf;
        char rawData[32]; /* not use */ 
        
    } param;
    
};

struct HttpLiveImplS
{
    HttpLiveT base;
    int sk_listen;
    int epfd;
    struct epoll_event *epEvents;

    uint16_t port;
    uint64_t size; /* 0 mean is live */
    char *path;
    MemPoolT *pool;

    enum HLStateE state;
    
    pthread_t tid;
    
    ZQueueT *msgQueue;
    ZListT bufList;
    int bufCacheSize;
    ZListT sessionList;
    int sessionNum;

    pthread_mutex_t lock;
};

static inline HLBufferT *HLBufferNew(MemPoolT *pool, char *data, int len)
{
    HLBufferT *buf = NULL;
    buf = Palloc(pool, sizeof(*buf));

    buf->data = Palloc(pool, len);
    memcpy(buf->data, data, len);
    buf->len = len;
    
    ZAtomicSet(&buf->ref, 1L);
    
    return buf;
}

static inline HLBufferT *HLBufferRefInc(HLBufferT *buf)
{
    ZAtomicInc(&buf->ref);
    return buf;
}

static inline void HLBufferRefDec(HLBufferT *buf, MemPoolT *pool)
{
    if (ZAtomicDec(&buf->ref) == 0)        
    {                                      
        Pfree(pool, buf->data);                  
        Pfree(pool, buf);                        
    }
    return ;
}

#define HLBufferDelete(buf, pool) HLBufferRefDec(buf, pool)

static int AsyncSockRecv(int sk, char *buf, int capability)
{
    int ret = 0, len = 0;
    
    while (1)
    {
        ret = recv(sk, buf + len, capability - len, MSG_DONTWAIT);
        if (ret < 0)
        {
            if (EAGAIN == errno || EWOULDBLOCK == errno)
            {
                /* noblocking socket */
                break;
            }
            else
            {
                len = -1;
                LogE("recv failure, errno(%d) ", errno);
                break;
            }
        }
        else if (ret == 0) /* close by peer...?? */
        {
            LogE("recv error, errno(%d)", errno);
            break;
        }
        else /* ret > 0 */
        {
            len += ret;
        }

        if (len == capability)
        {
            /* buf full */
            break;
        }
    }

    return len;
}

static int AsyncSockSend(int sk, const char *buf, int len)
{
    int ret = 0, done = 0;
    while (1)
    {
        ret = send(sk, buf + done, len - done, MSG_DONTWAIT);
        if (ret < 0)
        {
            if (EAGAIN == errno || EWOULDBLOCK == errno)
            {
                //buffer not ready
                break;
            }
            else
            {
                LogE("send failure, errno(%d)", errno);
                return -1;
            }
        }
        else if (ret == 0)
        {
            //buffer is full?
            break;
        }
        else // ret > 0
        {
            done += ret;
        }
        
        if (done == len)
        {
            /* finish... */
            break;
        }
    }
    
    return done;
}
 
/* only fill buffer into session buffer list, not send now */
static int SessionFill(struct HttpLiveImplS *impl, HLSessionT *session, HLBufferT *buf)
{
    HLBufferNodeT *bufNode = NULL;
    
    bufNode = Palloc(impl->pool, sizeof(*bufNode));
    bufNode->offset = 0;
    bufNode->buf = HLBufferRefInc(buf);
    ZListAddTail(&bufNode->node, &session->bufList);

    return 0;
}

static int SessionCreate(struct HttpLiveImplS *impl, int sk)
{
    HLSessionT *session = NULL;
    struct epoll_event event;
    int ret;

    LogI("a new session, sk(%d)", sk);
    session = Palloc(impl->pool, sizeof(*session));
    memset(session, 0x00, sizeof(*session));
    session->sk = sk;
    
    zsk_nonblocking(sk);
    session->state = HLSESSION_STATE_CONNECTED;
    ZListInit(&session->bufList);
    session->hdr = Palloc(impl->pool, 4096); /* store request header */
    session->hdrLen = 0;
    session->hdrValid = 0;
    
    event.data.ptr = session;
    event.events = EPOLLIN|EPOLLET|EPOLLONESHOT;  
    ret = epoll_ctl(impl->epfd, EPOLL_CTL_ADD, sk, &event);  
    if (ret == -1)
    {  
        LogE("epoll_ctl add failure, errno(%d)", errno);  
        return 0;
    }
    
    ZListAdd(&session->node, &impl->sessionList);
    impl->sessionNum++;

    return 0;
}

static void SessionDestroy(struct HttpLiveImplS *impl, HLSessionT *session)
{
    HLBufferNodeT *posBuf, *nextBuf;

	if (session->hdr != NULL)
	{
		Pfree(impl->pool,session->hdr);
		session->hdr = NULL;
	}
	
    ZListDel(&session->node);
    close(session->sk);
    session->sk = -1;

    ZListForEachEntrySafe(posBuf, nextBuf, &session->bufList, node)
    {
        HLBufferRefDec(posBuf->buf, impl->pool);
        ZListDel(&posBuf->node);
        Pfree(impl->pool, posBuf);
    }

	if (session != NULL)
	{
		Pfree(impl->pool,session);
		session = NULL;
	}
    
    return ;
}

char *ZStrstr(const char *str, const char *p)
{
    int strLen = strlen(str);
    int pLen = strlen(p);
    int i, j;

//    LogD("s(%d), p(%d)", strLen, pLen);
    
    int limit = strLen + 1 - pLen;
    for (i = 0; i < limit; i++)
    {
        for (j = 0; j < pLen; j++)
        {
   //         LogD("s: '%hhd', p: '%hhd'", str[i+j], p[j]);
            if (str[i+j] != p[j])
            {
                break;
            }
        }

        if ((j == pLen) && (str[i+j-1] == p[j-1]))
        {
            return (char *)&str[i];
        }
        
    }
    
    return NULL;
}

static int SessionTransProc(struct HttpLiveImplS *impl, struct epoll_event *env)
{
    HLSessionT *session = env->data.ptr;
    int ret;
    struct epoll_event event;
    
    if (session->state == HLSESSION_STATE_CONNECTED)
    {
        if (env->events & EPOLLIN)
        {
            int ret = AsyncSockRecv(session->sk, session->hdr + session->hdrLen, 
                                                 4096 - session->hdrLen);
            if (ret == -1)
            {
                LogW("recv failure, continue...");
                goto listen_read_event_out;
            }
            LogI("hdr:\n %s", session->hdr);        
            session->hdrLen += ret;
            static char *httpEndFlag = "\r\n\r\n"; 
            if (ZStrstr(session->hdr, httpEndFlag))
            {
                LogI("[HTTP] header reciver....");
                session->state = HLSESSION_STATE_RECVHDR;
                
                if (strncmp(session->hdr, "GET ", 4) == 0)
                {
                    session->hdrValid = 1; /* default invalid */
                    // TODO: get what, path if right...
                    session->pathValid = 1;
                }

                if (session->hdrValid && session->pathValid)
                {
                    session->response = response200;
                    session->responseLen = strlen(session->response);
                    
                }
                else if (session->hdrValid && !session->pathValid)
                {
                    session->response = response404;
                    session->responseLen = strlen(session->response);
                    
                } 
                else /* !session->hdrValid */
                {    
                    session->response = response503;
                    session->responseLen = strlen(session->response);
                }

                goto listen_write_event_out;
            }
            
        }
        goto listen_read_event_out; /* HLSESSION_STATE_CONNECTED */
        
    }
    else if (session->state == HLSESSION_STATE_RECVHDR)
    {
        if (env->events & EPOLLOUT)
        {            
            int ret;
            ret = AsyncSockSend(session->sk, session->response + session->responseSend, 
                                             session->responseLen - session->responseSend);
            session->responseSend += ret;

            if (session->responseSend == session->responseLen)
            {
               if (!session->pathValid || !session->hdrValid)
               {
                    goto close_session_out;
               }
               else
               {
                    LogI("[HTTP] response send.");
                    session->state = HLSESSION_STATE_TRANSFER;
                    goto listen_write_event_out;
               }
            }
        }
        goto listen_write_event_out;
    }
    else if (session->state == HLSESSION_STATE_TRANSFER)
    {
        if (env->events & EPOLLOUT)
        {
            HLBufferNodeT *pos, *next;
            if (ZListEmpty(&session->bufList))
            {
                LogD("session buffer list empty...");
                usleep(10000);
            }

            ZListForEachEntrySafe(pos, next, &session->bufList, node)
            {
                int ret;
                ret = AsyncSockSend(session->sk, pos->buf->data + pos->offset, pos->buf->len);
                if (ret == -1)
                {
                    LogE("socket send failure...");
                    break;
                }
                if (ret != pos->buf->len)
                {
                    pos->offset += ret;
                    /* nonblocking, wait next event... */
                    break;
                }
                /* this buffer nouse */
                ZListDel(&pos->node);
                HLBufferRefDec(pos->buf, impl->pool);
				Pfree(impl->pool, pos);
            }
        }
        goto listen_write_event_out;
    }
    else
    {
        Z_LOG_ASSERT(0, "should not be here...");
    }

listen_read_event_out:
    event.data.ptr = session;
    event.events = EPOLLIN|EPOLLET|EPOLLONESHOT;  
    ret = epoll_ctl(impl->epfd, EPOLL_CTL_DEL, session->sk, &event);  
    ret = epoll_ctl(impl->epfd, EPOLL_CTL_ADD, session->sk, &event);  
    if (ret != 0)
    {
        LogW("epoll ctl failure, sk(%d) errno(%d)", session->sk, errno);
    }
    return 0;
    
listen_write_event_out:
    event.data.ptr = session;
    event.events = EPOLLOUT|EPOLLET|EPOLLONESHOT;  
    ret = epoll_ctl(impl->epfd, EPOLL_CTL_DEL, session->sk, &event);  
    ret = epoll_ctl(impl->epfd, EPOLL_CTL_ADD, session->sk, &event);  
    if (ret != 0)
    {
        LogW("epoll ctl failure, sk(%d) errno(%d)", session->sk, errno);
    }
    return 0;

close_session_out:
    SessionDestroy(impl, session);
    return 0;
}


static int MsgProcFill(struct HttpLiveImplS *impl, HLBufferT *buf)
{
    HLBufferNodeT *posBuf, *nextBuf;
    HLBufferNodeT *newBuf;
    
    HLSessionT *posSession;

    newBuf = Palloc(impl->pool, sizeof(*newBuf));
    memset(newBuf, 0x00, sizeof(*newBuf));
    
    newBuf->buf = HLBufferRefInc(buf);
    newBuf->offset = 0;
    ZListAddTail(&newBuf->node, &impl->bufList);
    impl->bufCacheSize += newBuf->buf->len;
    
    if (!impl->sessionNum)
    {
        if (impl->bufCacheSize > (0x1<<21)) /* over 2MB, we will del 2 buffer */
        {
            LogW("cache full, remove some node...");
            posBuf = ZListFirstEntry(&impl->bufList, HLBufferNodeT, node);
            impl->bufCacheSize -= posBuf->buf->len;
            ZListDel(&posBuf->node);
            HLBufferRefDec(posBuf->buf, impl->pool);
            Pfree(impl->pool, posBuf);
        }
        /* not session, store buf in impl.bufList */
        return -1;
    }
    
    ZListForEachEntrySafe(posBuf, nextBuf, &impl->bufList, node)
    {
		char chunkSizeStr[32] = {0};
		sprintf(chunkSizeStr, CRLF"%x"CRLF, posBuf->buf->len);
		HLBufferT *chunkSizeBuf = HLBufferNew(impl->pool, chunkSizeStr, strlen(chunkSizeStr));

        ZListForEachEntry(posSession, &impl->sessionList, node)
        {
            SessionFill(impl, posSession, chunkSizeBuf);
            
            SessionFill(impl, posSession, posBuf->buf);
        }
        impl->bufCacheSize -= posBuf->buf->len;
        HLBufferRefDec(posBuf->buf, impl->pool);
        ZListDel(&posBuf->node);
        Pfree(impl->pool, posBuf);
		HLBufferRefDec(chunkSizeBuf, impl->pool);
    }

    return 0;
}

static int MsgProcListen(struct HttpLiveImplS *impl)
{
    int ret, i;
    ret = epoll_wait(impl->epfd, impl->epEvents, PERF_LIMIT, 100); /* 10 millisenconds */
    if (ret == -1)
    {
        LogE("epoll_wait error, errno(%d) ", errno);
        goto out;
    }
    else if (ret == 0)
    {
//        LogD("no any event...");
        /* no event, continue */
        goto out;
    }
    else if (ret < 0)
    {
        Z_LOG_ASSERT(0, "should not be here...");
        goto out;
    }

    /* events */
    for (i = 0; i < ret; i++)
    {
        if (impl->epEvents[i].events & (EPOLLERR|EPOLLHUP))
        {
            LogE("epoll error: '%x' ('%x'%x')", impl->epEvents[i].events, EPOLLERR, EPOLLHUP);
            if (impl->epEvents[i].data.fd != impl->sk_listen)
            {
                HLSessionT *session = impl->epEvents[i].data.ptr;
                SessionDestroy(impl, session);
            }
            continue;
        }

        if (impl->epEvents[i].data.fd == impl->sk_listen)
        {
            while (1)  
            {  
                struct sockaddr in_addr;  
                socklen_t in_len;  
                int infd;  
  
                in_len = sizeof(in_addr);  
                infd = accept(impl->sk_listen, &in_addr, &in_len);  
                if (infd == -1)
                {  
                    if ((errno == EAGAIN) || (errno == EWOULDBLOCK))  
                    {  
                        /* We have processed all incoming connections. */  
                        break;
                    }
                    else
                    {
                        LogE("accept error... errno(%d) ", errno);  
                        break;
                    }
                }  
                /* a new connection */
                SessionCreate(impl, infd);
            }
            
        }
        else /* connection */
        {
            SessionTransProc(impl, &impl->epEvents[i]);
        }
        
    }

out:
    return 0;
}

static int HLMsgQueueClear(struct HttpLiveImplS *impl)
{
    HLMessageT *msg = NULL;
    while ((msg = ZQueuePop(impl->msgQueue)) != NULL)
    {
        if (msg->type == HL_MSG_FILL)
        {
            HLBufferDelete(msg->param.buf, impl->pool);
        }

        Pfree(impl->pool, msg);
    }
    
    return 0;
}

static inline int LightMsgSend(struct HttpLiveImplS *impl, enum HLMsgTypeE type)
{
    HLMessageT *msg = NULL;
    
    msg = Palloc(impl->pool, sizeof(*msg));    
    msg->type = type;
    ZQueuePush(impl->msgQueue, msg);

    return 0;
}

static void *MessageProcessCore(void *arg)
{
    struct HttpLiveImplS *impl = NULL;
    HLMessageT *msg;
    impl = arg;
    
    while (1)
    {
        msg = ZQueuePop(impl->msgQueue);
        if (!msg)
        {
            /* should not be here unless exit... */
            usleep(100000); 
            continue;
        }

        switch (msg->type)
        {
        case HL_MSG_STOP:
        {
            LogI("HttpLive srv stop...");
            /* should exit this thread, when stop... */
            Pfree(impl->pool, msg);
            return NULL;
        }
        
        case HL_MSG_FILL:
        {
//            LogD("----------");
            MsgProcFill(impl, msg->param.buf);

            HLBufferRefDec(msg->param.buf, impl->pool);
            Pfree(impl->pool, msg);            
            break;
        }
        
        case HL_MSG_LISTEN:
        {
            MsgProcListen(impl);
            ZQueuePush(impl->msgQueue, msg);
            break;
        }            
        default:
            Z_LOG_ASSERT(0, "should not be here...");
            break;
        }
        
    }
}

int __HLStart(HttpLiveT *hl)
{
    struct HttpLiveImplS *impl;
//    HLMessageT *msg = NULL;
    int ret;
    Z_ASSERT(hl);
    
    impl = ZContainerOf(hl, struct HttpLiveImplS, base);
    pthread_mutex_lock(&impl->lock);
    
    if (impl->state == HL_STATE_RUNNING)
    {
        LogW("already statred...");
        return 0;
    }

    ret = pthread_create(&impl->tid, NULL, MessageProcessCore, impl);
    if (ret != 0)
    {
        LogE("thread init failure, errno(%d).", errno);
        return -1;
    }

    struct epoll_event env;
    env.data.fd = impl->sk_listen;
    env.events = EPOLLIN | EPOLLET;
    epoll_ctl(impl->epfd, EPOLL_CTL_ADD, impl->sk_listen, &env);

    LightMsgSend(impl, HL_MSG_LISTEN);

    impl->state = HL_STATE_RUNNING;
    
    pthread_mutex_unlock(&impl->lock);
    return 0;
}

int32_t __HLFill(HttpLiveT *hl, char *data, int len)
{
    struct HttpLiveImplS *impl = NULL;
    HLMessageT *msg = NULL;
    Z_ASSERT(hl);
    
    impl = ZContainerOf(hl, struct HttpLiveImplS, base);

    if (impl->state != HL_STATE_RUNNING)
    {
        LogE("proxy not running...");
        return -1;
    }
    
    if (impl->bufCacheSize > (0x1<<20)) /* more than 1MB, wait some time */
    {
        usleep(100000);
    }

    msg = Palloc(impl->pool, sizeof(*msg));
    
    msg->type = HL_MSG_FILL;
    msg->param.buf = HLBufferNew(impl->pool, data, len);;
//    LogD("fill data, size(%d)", len);
    ZQueuePush(impl->msgQueue, msg);
    
    return 0;
}

int32_t __HLStop(HttpLiveT *hl)
{
    struct HttpLiveImplS *impl;

    Z_ASSERT(hl);
    
    impl = ZContainerOf(hl, struct HttpLiveImplS, base);
    pthread_mutex_lock(&impl->lock);

    if (impl->state == HL_STATE_STOPED)
    {
        /* already stop now */
        return 0;
    }
    impl->state = HL_STATE_STOPED;
    
    pthread_mutex_unlock(&impl->lock);

    HLMsgQueueClear(impl);

    LightMsgSend(impl, HL_MSG_STOP);
    
    struct epoll_event env;
    epoll_ctl(impl->epfd, EPOLL_CTL_DEL, impl->sk_listen, &env); 
    /* before 2.6.9 request a non-null env*/

    pthread_join(impl->tid, NULL);
    return 0;
}

struct HttpLiveOpsS hlOps =
{
    .start = __HLStart,
    .fill = __HLFill,
    .stop = __HLStop
};

HttpLiveT *HLCreate(uint16_t port, uint64_t size, char *path)
{
    struct HttpLiveImplS *impl;
    impl = malloc(sizeof(*impl));
    int ret;
    struct sockaddr_in saddr;
    
    if (!impl)
    {
        LogE("malloc failure, errno(%d), size(%d)", errno, sizeof(*impl));
        goto failure;
    }
    
    memset(impl, 0x00, sizeof(*impl));
    impl->pool = MemPoolCreate(NULL);
    if (!impl)
    {
        LogE("malloc failure, errno(%d), size(%d)", errno, sizeof(*impl));
        goto failure;
    }

    impl->base.ops = &hlOps;
    impl->sk_listen = -1;
    impl->port = port;
    impl->size = size;
    impl->path = Pstrdup(impl->pool, path);
    pthread_mutex_init(&impl->lock, NULL);
    
    impl->sk_listen = socket(AF_INET, SOCK_STREAM, 0);
    if (impl->sk_listen == -1)
    {
        LogE("socket create failure, errno(%d)", errno);
        goto failure;
    }
    LogI(" listen sk(%d)", impl->sk_listen);
    
    ret = zsk_nonblocking(impl->sk_listen);
    if (ret != 0)
    {
        LogE("nonblock set failure. errno(%d)", errno);
        goto failure;
    }
    
    saddr.sin_family = AF_INET;
    saddr.sin_port = htons(impl->port);
    saddr.sin_addr.s_addr = htonl(INADDR_ANY);
    ret = bind(impl->sk_listen, (struct sockaddr *)&saddr, sizeof(saddr));
    if (ret != 0)
    {
        LogE("bind failure, errno(%d)", errno);
        goto failure;
    }

    /* listen */
    ret = listen(impl->sk_listen, PERF_LIMIT);
    if (ret != 0)
    {
        LogE("listen failure, errno(%d)", errno);
        goto failure;
    }

    impl->epfd = epoll_create(PERF_LIMIT);

    impl->epEvents = Palloc(impl->pool, sizeof(impl->epEvents) * 1024);

    impl->msgQueue = ZQueueCreate(impl->pool);
    Z_ASSERT(impl->msgQueue != NULL);
    
    ZListInit(&impl->bufList);
    ZListInit(&impl->sessionList);
    impl->state = HL_STATE_STOPED;
    impl->bufCacheSize = 0;
    impl->sessionNum = 0;
    
    return &impl->base;

failure:

    if (impl)
    {
        if (impl->path)
        {
            Pfree(impl->pool, impl->path);
            impl->path = NULL;
        }
    
        if (impl->sk_listen >= 0)
        {
            close(impl->sk_listen);
            impl->sk_listen = -1;
        }
        
        if (impl->pool)
        {
            MemPoolDestroy(impl->pool);
            impl->pool = NULL;
        }

        free(impl);
    }

    return NULL;
}

void HLDestroy(HttpLiveT *hl)
{
    struct HttpLiveImplS *impl;

    Z_ASSERT(hl);
    
    impl = ZContainerOf(hl, struct HttpLiveImplS, base);
    
    if (impl->path)
    {
        Pfree(impl->pool, impl->path);
        impl->path = NULL;
    }

	HLMsgQueueClear(impl);
	ZQueueDestroy(impl->msgQueue);

    if (impl->epEvents)
    {
        Pfree(impl->pool, impl->epEvents);
        impl->epEvents = NULL;
    }	
    
    if (impl->sk_listen >= 0)
    {
        close(impl->sk_listen);
        impl->sk_listen = -1;
    }
    
    if (impl->pool)
    {
        MemPoolDestroy(impl->pool);
        impl->pool = NULL;
    }
    
    free(impl);
    return ;
}

